package com.idanchuang.component.sequence.id.service;

import com.idanchuang.component.base.JsonResult;
import com.idanchuang.component.base.exception.common.ErrorCode;
import com.idanchuang.component.base.exception.core.ExFactory;
import com.idanchuang.component.logback.helper.LogContext;
import com.idanchuang.component.sequence.id.config.SequenceConfigBean;
import com.idanchuang.sequence.service.api.client.ISequenceFeignClient;
import com.idanchuang.sequence.service.api.dto.BatchIdDTO;
import feign.RetryableException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 默认ID序列实现
 */
public class GeneralSeqIdServiceImpl implements GeneralSeqIdService {

    private static final Logger log = LoggerFactory.getLogger(GeneralSeqIdServiceImpl.class);

    private final ISequenceFeignClient iSequenceFeignClient;
    private final SequenceConfigBean sequenceConfigBean;
    private final Map<String, LocalQueue> localQueueMap;
    private final ExecutorService apply_worker;

    public GeneralSeqIdServiceImpl(ISequenceFeignClient iSequenceFeignClient, SequenceConfigBean sequenceConfigBean) {
        this.iSequenceFeignClient = iSequenceFeignClient;
        this.sequenceConfigBean = sequenceConfigBean;
        this.localQueueMap = new HashMap<>();
        this.apply_worker = new ThreadPoolExecutor(sequenceConfigBean.getCorePoolSize(), sequenceConfigBean.getMaximumPoolSize(),
                sequenceConfigBean.getWorkKeepAliveTime(), TimeUnit.SECONDS, new LinkedBlockingQueue<>(sequenceConfigBean.getWorkQueueSize()),
                (runnable) -> new Thread(runnable, "apply-worker"),
                (r, executor) -> {
                    LogContext.setKeyword("sequence_id_error");
                    log.error("apply-worker is busy, use current thread, config: {}", sequenceConfigBean);
                    r.run();
                });
    }

    @Override
    public long nextId(String appId, String tableName) {
        return nextId(appId, tableName, 1).get(0);
    }

    @Override
    public List<Long> nextId(String appId, String tableName, int count) {
        if (count < 1 || count > sequenceConfigBean.getMaxBatchCount()) {
            throw ExFactory.throwCommon(ErrorCode.PARAM_ERROR, "nextId > batchCount must be gt 0 and lt maxBatchCount[" + sequenceConfigBean.getMaxBatchCount() + "]");
        }
        LocalQueue localQueue = getLocalQueue(appId, tableName, count);
        List<Long> list = new ArrayList<>(count);
        while (--count >= 0) {
            list.add(localQueue.poll());
        }
        return list;
    }

    @Override
    public long resetId(String appId, String tableName, long initId) {
        try {
            iSequenceFeignClient.resetId(appId, tableName, initId).mustSuccess();
        } catch (RetryableException e) {
            if (!sequenceConfigBean.isEnableRetryRpc()) {
                throw e;
            }
            iSequenceFeignClient.resetId(appId, tableName, initId).mustSuccess();
        }
        clearLocalQueue(appId, tableName);
        return initId;
    }

    @Override
    public synchronized void setMinId(String appId, String tableName, long minId) {
        long nextId = iSequenceFeignClient.nextId(appId, tableName).getData();
        if (nextId < minId) {
            iSequenceFeignClient.resetId(appId, tableName, minId - 1);
        }
    }

    /**
     * 清空当前缓存的id集合
     * @param appId appid
     * @param tableName 表名
     */
    private void clearLocalQueue(String appId, String tableName) {
        String localKey = getLocalKey(appId, tableName);
        LocalQueue localQueue = localQueueMap.get(localKey);
        if (localQueue != null) {
            localQueue.currentQueue.clear();
            localQueue.readyQueue.clear();
            localQueue.fillReady();
        }
    }

    private LocalQueue getLocalQueue(String appId, String tableName, int forCount) {
        String localKey = getLocalKey(appId, tableName);
        LocalQueue localQueue = localQueueMap.get(localKey);
        if (localQueue == null) {
            // 初始化LocalQueue
            localQueue = new LocalQueue(appId, tableName);
            localQueueMap.put(localKey, localQueue);
        }
        localQueue.setCurrentPollCount(forCount);
        return localQueue;
    }

    private String getLocalKey(String appId, String tableName) {
        return "local_" + appId + "_" + tableName;
    }

    /**
     * 本地内存ID缓冲队列
     */
    private class LocalQueue {

        /** 记录客户端最后一次获取ID的批量数, 在大数额申请id时可以减少向sequence服务申请次数, 提升性能 */
        private Integer currentPollCount;
        private final String appId;
        private final String tableName;
        /** 是否正在向sequence服务申请中 */
        private boolean applying = false;
        /** 当前ID队列, 如果此队列取空, 则将readyQueue加载至此队列 */
        private final LinkedBlockingQueue<Long> currentQueue = new LinkedBlockingQueue<>();
        /** 预备中的ID队列, 当队列上位后, 会重新向sequence服务申请一批ID作为预备ID */
        private final LinkedBlockingQueue<Long> readyQueue = new LinkedBlockingQueue<>();

        private LocalQueue(String appId, String tableName) {
            this.appId = appId;
            this.tableName = tableName;
            init();
        }

        private synchronized void init() {
            log.debug("init new LocalQueue, appId: {}, tableName: {}", appId, tableName);
            fillReady();
        }

        private synchronized Long poll() {
            exchangeQueueIfNecessary();
            return currentQueue.poll();
        }

        private void exchangeQueueIfNecessary() {
            if (!currentQueue.isEmpty()) {
                return;
            }
            synchronized (readyQueue) {
                try {
                    if (readyQueue.isEmpty()) {
                        if (!applying) {
                            fillReady();
                        }
                        log.debug("readyQueue is unfilled, waiting apply, appId: {}, tableName: {}", appId, tableName);
                        readyQueue.wait();
                        log.debug("readyQueue is filled, go on, appId: {}, tableName: {}", appId, tableName);
                    }
                    currentQueue.addAll(readyQueue);
                    readyQueue.clear();
                    fillReady();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }

        private void fillReady() {
            if (!applying) {
                applying = true;
                log.debug("fileReady > appId: {}, tableName: {}", appId, tableName);
                apply_worker.submit(() -> {
                    try {
                        synchronized (readyQueue) {
                            int count = sequenceConfigBean.getLocalQueueSize();
                            if (currentPollCount != null) {
                                count = Math.max(currentPollCount, count);
                            }
                            JsonResult<BatchIdDTO> result;
                            try {
                                result = iSequenceFeignClient.nextBatchId(appId, tableName, count);
                            } catch (RetryableException e) {
                                if (!sequenceConfigBean.isEnableRetryRpc()) {
                                    throw e;
                                }
                                log.warn("iSequenceFeignClient.nextBatchId({}, {}, {}) failed, will retry once.",
                                        appId, tableName, count);
                                // rpc 异常, 自动重来一次
                                result = iSequenceFeignClient.nextBatchId(appId, tableName, count);
                                // 打印重试结果
                                log.info("iSequenceFeignClient.nextBatchId({}, {}, {}) retry result: {}",
                                        appId, tableName, count, result);
                            }
                            result.mustSuccess();
                            BatchIdDTO dto = result.getData();
                            readyQueue.addAll(dto.getIdList());
                            readyQueue.notify();
                            log.debug("fileReady ok > appId: {}, tableName: {}", appId, tableName);
                            applying = false;
                        }
                    } catch (Exception e) {
                        log.error("fillReady error, will retry after 1 seconds", e);
                        applying = false;
                        try {
                            Thread.sleep(1000L);
                        } catch (InterruptedException ex) {
                            ex.printStackTrace();
                        }
                        fillReady();
                    }
                });
            }
        }

        public Integer getCurrentPollCount() {
            return currentPollCount;
        }

        public void setCurrentPollCount(Integer currentPollCount) {
            this.currentPollCount = currentPollCount;
        }
    }

}
